package com.sundata.practice;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;

public class AkkaDAG {
    static class TaskActor extends AbstractActor {
        private final String taskName;
        private final ActorRef[] dependencies;

        public TaskActor(String taskName, ActorRef... dependencies) {
            this.taskName = taskName;
            this.dependencies = dependencies;
        }

        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .matchEquals("run", msg -> {
                        for (ActorRef dep : dependencies) {
                            dep.tell("run", getSelf());
                        }
                        System.out.println("Task " + taskName);
                    })
                    .build();
        }
    }

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("DAGSystem");

        ActorRef d = system.actorOf(Props.create(TaskActor.class, "D"), "D");
        ActorRef b = system.actorOf(Props.create(TaskActor.class, "B", d), "B");
        ActorRef c = system.actorOf(Props.create(TaskActor.class, "C", d), "C");
        ActorRef a = system.actorOf(Props.create(TaskActor.class, "A", b, c), "A");

        a.tell("run", ActorRef.noSender());

        system.terminate();
    }
}